package com.toutiao.nssue

import scala.collection.mutable.ArrayBuffer

import org.elasticsearch.common.settings.{Settings,ImmutableSettings}
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.elasticsearch.index.query.FilterBuilders._
import org.elasticsearch.index.query.QueryBuilders._
import org.elasticsearch.action.search.SearchType
import org.elasticsearch.common.unit.TimeValue
import org.elasticsearch.search.aggregations.metrics.max.Max
import org.elasticsearch.search.aggregations.AggregationBuilders.max

import net.liftweb.json.{JsonAST,Extraction,Printer,parse}
import net.liftweb.json.JsonDSL._

import scala.collection.JavaConversions._

import java.util.Date
import java.sql.Timestamp

import org.apache.log4j.Logger

import com.toutiao.issue.SimpleDate


class DocStore(
        esName:String,
        esUri:String,
        esPort:Int,
        esIndexName:String,
        esTypeName:String
    ) {
  
  private val log = Logger.getLogger(this.getClass.getName)
  log.info("ES_NAME "  +esName)
  log.info("ES_URI "   +esUri)
  log.info("ES_PORT "  +esPort)
  log.info("ES_INDEX " +esIndexName)
  log.info("ES_TYPE "  +esTypeName)
  
  private val s = ImmutableSettings.settingsBuilder.
                put("cluster.name",esName).build
  
  private val ES = new TransportClient(s).addTransportAddress(
		  	new InetSocketTransportAddress(esUri,esPort))
	
	implicit val _ = net.liftweb.json.DefaultFormats
	
  def getMaxTime(tt:String) = {
    val resp = ES.prepareSearch(esIndexName).setTypes(esTypeName)
    			.addAggregation(max("maxTime").field(tt)).execute.actionGet
    val r:Max = resp.getAggregations.get("maxTime")
    val x = r.getValue
    val z = new Date(new Timestamp(x.toLong).getTime)
    val t = SimpleDate.increment(z, -8*60) // 这里不知道会不会有问题
    t
  }

  def retrieve(st: Date, et: Date) = {
    val rt = "create_time"
    val xs = new ArrayBuffer[RawDocument]
    var scrollResp = ES.prepareSearch(esIndexName).
      setTypes(esTypeName).
      setSearchType(SearchType.SCAN).
      setScroll(new TimeValue(60000)).
      setPostFilter(rangeFilter(rt).
        from(SimpleDate.format(st)).
        to(SimpleDate.format(et)).
        includeLower(true).
        includeUpper(false)).
      setSize(200).
      execute().actionGet()
    var z = 1
    while (z > 0) {
      scrollResp = ES.prepareSearchScroll(scrollResp.getScrollId).
        setScroll(new TimeValue(60000)).
        execute.actionGet
      for (n <- scrollResp.getHits.hits) {
        val s = n.getSource
        val name = s.get("id").toString
        val t = SimpleDate.parse(s.get("create_time").asInstanceOf[String])
        val title = s.get("cut_title").asInstanceOf[String]
        val content = s.get("cut_content").asInstanceOf[String]
        val vs =
          try {
            parse(s.get("term_score").asInstanceOf[String]).extract[Map[String, Double]].
              map(vv => {
                val v = vv._1.replace(":", "")
                (v, vv._2)
              }) //.filter(_._2 > 5.0)
          } catch {
            case _: Throwable => Map[String, Double]()
          }
        val u = s.get("url").asInstanceOf[String]
        xs += new RawDocument(name,title,content,vs,u,t)

      }

      if (scrollResp.getHits.hits.length == 0)
        z = 0
    }

    //_log.info("Got "+xs.size)
    xs.toArray.sortWith(_.date.getTime < _.date.getTime)
  }
  
  def close = ES.close
  
}